package com.hooray.elasticsearch;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.sort.SortBuilder;

import com.hooray.elasticsearch.exception.ElasticSearchException;
import com.hooray.elasticsearch.util.FastJsonUtil;

/**
 * 
 * @Desc Elasticsearch Client
 *
 * @Author daijunjie
 * @DateTime 2016年4月14日 下午4:45:24
 *
 */
public class ElasticsearchClient {
	public static final int DEFAULT_PORT = 9300;
	public static final int DEFAULT_BULK_SIZE = 3000;
	private String clusterName;
	private String hosts;
	private Client client;
	private volatile boolean started;
	private int bulkSize = DEFAULT_BULK_SIZE;

	public ElasticsearchClient(String clusterName, String hosts) {
		this.clusterName = clusterName;
		this.hosts = hosts;
	}

	public void start() {
		synchronized (this) {
			if (!started) {
				openClient(this.clusterName, this.hosts);
				started = true;
			}
		}
	}

	public void stop() {
		synchronized (this) {
			if (started) {
				closeClient();
				started = false;
			}
		}
	}

	protected void openClient(String clusterName, String hosts) {
		Settings settings = Settings.settingsBuilder()
		/** 集群名称 */
		.put("cluster.name", clusterName)
		/** 发现集群中的所有节点 */
		.put("client.transport.sniff", true)
		/** 构建 */
		.build();
		TransportClient transportClient = TransportClient.builder().settings(settings).build();
		for (String host : hosts.trim().split(",")) {
			String[] hostnameAndPort = host.split(":");
			String hostname = hostnameAndPort[0].trim();
			int port = hostnameAndPort.length == 2 ? Integer.parseInt(hostnameAndPort[1].trim()) : DEFAULT_PORT;
			transportClient.addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress(hostname, port)));
		}
		closeClient();
		client = transportClient;
	}

	protected void closeClient() {
		if (client != null) {
			client.close();
			client = null;
		}
	}

	/**
	 * 创建索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param id
	 *            文档ID，可以为null，当传null时，ES会自动创建
	 * @param jsonSource
	 *            json格式文档source
	 * @return
	 */
	public IndexResponse createIndex(String index, String type, String id, String jsonSource) {
		return client.prepareIndex(index, type, id).setSource(jsonSource).get();
	}

	/**
	 * 创建索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param id
	 *            文档ID，可以为null，当传null时，ES会自动创建
	 * @param sourceBuilder
	 * 
	 * @return
	 */
	public IndexResponse createIndex(String index, String type, String id, XContentBuilder sourceBuilder) {
		return client.prepareIndex(index, type, id).setSource(sourceBuilder).get();
	}

	/**
	 * 创建索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param id
	 *            文档ID，可以为null，当传null时，ES会自动创建
	 * @param mapSource
	 *            map格式文档source
	 * @return
	 */
	public IndexResponse createIndex(String index, String type, String id, Map<String, ?> mapSource) {
		return client.prepareIndex(index, type, id).setSource(mapSource).get();
	}

	/**
	 * 创建索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param id
	 *            文档ID，可以为null，当传null时，ES会自动创建
	 * @param beanSource
	 *            bean格式文档source，不能为String类型
	 * @return
	 */
	public IndexResponse createIndex(String index, String type, String id, Object beanSource) {
		return client.prepareIndex(index, type, id).setSource(FastJsonUtil.toJson(beanSource)).get();
	}

	/**
	 * 获取索引
	 * 
	 * @param index
	 * @param type
	 * @param id
	 * @return
	 */
	public GetResponse getIndex(String index, String type, String id) {
		// 设置OperationThreaded为false，表示在同一线程返回
		GetResponse response = client.prepareGet(index, type, id).setOperationThreaded(false).get();
		return response;
	}

	/**
	 * 删除索引
	 * 
	 * <p>
	 * 如果传入的索引中，有一个不存在，将导致所有删除失败
	 * </p>
	 * 
	 * @param client
	 * @param indices
	 * @return
	 */
	public boolean deleteIndexs(String... indices) {
		DeleteIndexRequest delete = new DeleteIndexRequest(indices);
		ActionFuture<DeleteIndexResponse> future = client.admin().indices().delete(delete);
		DeleteIndexResponse response = future.actionGet();
		return response.isAcknowledged();
	}

	/**
	 * 删除索引下指定ID的文档
	 * 
	 * @param client
	 * @param index
	 * @param type
	 * @param id
	 * @return
	 */
	public boolean deleteIndex(String index, String type, String id) {
		DeleteResponse dresponse = client.prepareDelete(index, type, id).get();
		return dresponse.isFound();
	}

	/**
	 * 更新文档
	 * 
	 * @param client
	 * @param index
	 * @param type
	 * @param id
	 * @param param
	 * @return
	 * @throws IOException
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public static UpdateResponse updateIndex(Client client, String index, String type, String id, Map<String, Object> param) throws IOException, InterruptedException, ExecutionException {
		UpdateResponse response = client.prepareUpdate(index, type, id).setDoc(param).get();
		return response;
	}

	/**
	 * 添加或更新文档
	 * 
	 * @param client
	 * @param index
	 * @param type
	 * @param id
	 * @param param
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public static UpdateResponse upsertIndex(Client client, String index, String type, String id, Map<String, Object> param) throws InterruptedException, ExecutionException {
		IndexRequest indexRequest = new IndexRequest(index, type, id);
		UpdateRequest updateRequest = new UpdateRequest(index, type, id).doc(param).upsert(indexRequest);
		UpdateResponse response = client.update(updateRequest).get();
		return response;
	}

	/**
	 * 获取多个文档
	 * 
	 * @param index
	 * @param type
	 * @param ids
	 * @return
	 */
	public MultiGetResponse multiGetIndex(String index, String type, String... ids) {
		MultiGetResponse multiGetItemResponses = client.prepareMultiGet().add(index, type, ids).get();
		MultiGetItemResponse[] items = multiGetItemResponses.getResponses();
		if (items != null) {
			for (MultiGetItemResponse item : items) {
				if (!item.isFailed()) {
					item.getResponse().getSourceAsBytes();
				}
			}
		}
		return multiGetItemResponses;
	}

	/**
	 * 批量添加索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param sourceList
	 *            对象支持Map<String, ?>,JSON
	 *            String,XContentBuilder,JavaBean(get,set method)
	 * @throws ElasticSearchException
	 */
	@SuppressWarnings("unchecked")
	public BulkResponse bulkAddIndex(String index, String type, List<Object> sourceList) {
		BulkRequestBuilder bulkRequest = client.prepareBulk();
		if (sourceList.size() > bulkSize) {
			throw new ElasticSearchException("sourceList size can not greater than " + bulkSize);
		}
		for (Object source : sourceList) {
			IndexRequestBuilder indexRequest = client.prepareIndex(index, type);
			if (source instanceof Map) {
				indexRequest.setSource((Map<String, ?>) source);
			} else if (source instanceof String) {
				indexRequest.setSource((String) source);
			} else if (source instanceof XContentBuilder) {
				indexRequest.setSource((XContentBuilder) source);
			} else {
				indexRequest.setSource(FastJsonUtil.toJson(source));
			}
			bulkRequest.add(indexRequest);
		}
		BulkResponse bulkResponse = bulkRequest.execute().actionGet();
		return bulkResponse;
	}

	/**
	 * 批量添加索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param sourceList
	 *            对象支持Map<String, ?>,JSON String,JavaBean(get,set method)
	 * @throws ElasticSearchException
	 */
	public BulkResponse bulkAddIndex(List<IndexRequestBuilder> indexList) throws ElasticSearchException {
		BulkRequestBuilder bulkRequest = client.prepareBulk();
		for (IndexRequestBuilder indexRequest : indexList) {
			bulkRequest.add(indexRequest);
		}
		BulkResponse bulkResponse = bulkRequest.execute().actionGet();
		if (bulkResponse.hasFailures()) {
			throw new ElasticSearchException(bulkResponse.buildFailureMessage());
		}
		return bulkResponse;
	}

	/**
	 * 搜索
	 * 
	 * @param index
	 * @param type
	 * @param searchType
	 * @param queryBuilder
	 * @param filterBuilder
	 * @param sortBuilder
	 * @param aggregationBuilder
	 * @param from
	 * @param size
	 * @return
	 * @throws IOException
	 */
	public SearchResponse search(String index, String type, SearchType searchType, QueryBuilder queryBuilder, QueryBuilder filterBuilder, SortBuilder sortBuilder, AggregationBuilder<?> aggregationBuilder, int from, int size) {
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index);
		if (StringUtils.isNotEmpty(type)) {
			searchRequestBuilder.setTypes(type);
		}
		if (searchType != null) {
			searchRequestBuilder.setSearchType(searchType);
		}
		if (queryBuilder != null) {
			searchRequestBuilder.setQuery(queryBuilder);
		}
		if (filterBuilder != null) {
			searchRequestBuilder.setPostFilter(filterBuilder);
		}
		if (sortBuilder != null) {
			searchRequestBuilder.addSort(sortBuilder);
		}
		if (from > 0) {
			searchRequestBuilder.setFrom(from);
		}
		if (size > 0) {
			searchRequestBuilder.setSize(size);
		}
		if (aggregationBuilder != null) {
			searchRequestBuilder.addAggregation(aggregationBuilder);
		}
		searchRequestBuilder.setExplain(true);
		SearchResponse response = searchRequestBuilder.execute().actionGet();
		return response;
	}

	public long count(String index, QueryBuilder queryBuilder) {
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index);
		if (queryBuilder != null) {
			searchRequestBuilder.setQuery(queryBuilder);
		}
		SearchResponse response = searchRequestBuilder.setSize(0).execute().actionGet();
		return response.getHits().getTotalHits();
	}

	public SearchResponse addAggregation(SearchRequestBuilder searchRequestBuilder, AggregationBuilder<?> aggregationBuilder) {
		SearchResponse response = searchRequestBuilder.addAggregation(aggregationBuilder).execute().actionGet();
		return response;
	}

	public static XContentBuilder xContentBuilder() throws IOException {
		return jsonBuilder();
	}

	public Client getClient() {
		return client;
	}

	public void setClient(Client client) {
		this.client = client;
	}

	public int getBulkSize() {
		return bulkSize;
	}

	public void setBulkSize(int bulkSize) {
		this.bulkSize = bulkSize;
	}

}
